/* Copyright (c) 2023 Renmin University of China
RMDB is licensed under Mulan PSL v2.
You can use this software according to the terms and conditions of the Mulan PSL v2.
You may obtain a copy of Mulan PSL v2 at:
        http://license.coscl.org.cn/MulanPSL2
THIS SOFTWARE IS PROVIDED ON AN "AS IS" BASIS, WITHOUT WARRANTIES OF ANY KIND,
EITHER EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO NON-INFRINGEMENT,
MERCHANTABILITY OR FIT FOR A PARTICULAR PURPOSE.
See the Mulan PSL v2 for more details. */

#pragma once
#include "common/common.h"
#include "defs.h"
#include "errors.h"
#include "executor_abstract.h"
#include <cstddef>
#include <cstdio>
#include <cstring>
#include <vector>
#include <string>

#include <iostream>
#include <fstream>
#include <sys/mman.h>
#include <sys/stat.h>
#include <fcntl.h>
#include <unistd.h>
#include <sstream>

class LoadExecutor : public AbstractExecutor {
   private:
    TabMeta tab_;                   // 表的元数据
    std::string file_path_;         // 需要读取的数据文件路径
    RmFileHandle *fh_;              // 表的数据文件句柄
    std::string tab_name_;          // 表名称
    Rid rid_;                       // 插入的位置，由于系统默认插入时不指定位置，因此当前rid_在插入后才赋值
    SmManager *sm_manager_;

    std::vector<Value> values_;

   public:
    LoadExecutor(SmManager *sm_manager, const std::string &tab_name, const std::string &file_path, Context *context) {
        sm_manager_ = sm_manager;
        tab_ = sm_manager_->db_.get_table(tab_name);
        file_path_ = file_path;
        tab_name_ = tab_name;
        fh_ = sm_manager_->fhs_.at(tab_name).get();
        context_ = context;
    };

    ~LoadExecutor() {
    }
    std::unique_ptr<RmRecord> Next() override {
        
        // context_->lock_mgr_->lock_exclusive_on_table(context_->txn_, fh_->GetFd());
        readCSV(file_path_);

        return nullptr;
    }

    void readCSV(const std::string& filename) {
        std::chrono::high_resolution_clock::time_point start = std::chrono::high_resolution_clock::now();

        int fd = open(filename.c_str(), O_RDONLY);
        if(fd == -1){
            throw FileNotFoundError(filename);
        }

        struct stat sb;
        if (fstat(fd, &sb) == -1) {
            std::cerr << "Failed to get file size: " << filename << std::endl;
            throw RMDBError("Failed to get file size");
        }
        size_t fileSize = sb.st_size;

        char* data = static_cast<char*>(mmap(nullptr, fileSize, PROT_READ, MAP_PRIVATE, fd, 0));
        if (data == MAP_FAILED) {
            data = nullptr;
            throw RMDBError("Failed to map file");
        }

        close(fd);

        int file_point = 0;
        while(data[file_point] != '\n') {
            file_point++;
        }
        file_point+=1;

        Value val;
        RmRecord rec(fh_->get_file_hdr().record_size);
        Rid rid;
        int tmp_int;
        float tmp_float;
        while(file_point < fileSize) {
            std::string line;
            char* lineStart = data + file_point;
            for (size_t i = file_point; i < fileSize; ++i) {
                if (data[i] == '\n') {
                    line = std::string(lineStart, &data[i]);
                    break;
                }
            }
            file_point += line.length() + 1;
            std::stringstream ss(line);

            std::vector<std::string> tokens;
            std::string token;
            while (std::getline(ss, token, ',')) {
                tokens.push_back(token);
            }


            for(int i = 0; i < tab_.cols.size(); i++){
                auto& col = tab_.cols[i];
                switch (col.type) {
                    case TYPE_INT:
                        tmp_int = std::stoi(tokens[i]);
                        memcpy(rec.data + col.offset, &tmp_int, col.len);
                        // std::cout << "int = "<< val.int_val << std::endl;
                        break;
                    case TYPE_FLOAT:
                        tmp_float = std::stof(tokens[i]);
                        memcpy(rec.data + col.offset, &tmp_float, col.len);
                        // std::cout << "float = "<< val.float_val << std::endl;
                        break;
                    case TYPE_STRING:
                        memcpy(rec.data + col.offset, tokens[i].c_str(), col.len);
                        // std::cout << "string = "<< val.str_val << std::endl;
                        break;
                    default:
                        throw RMDBError("Unsupported type");
                }
            }

            size_t num_indexes = tab_.indexes.size();
            // 插入数据
            rid = fh_->insert_record(rec.data, context_, true);

            // 插入索引
            // Insert into index
            for(size_t i = 0; i < num_indexes; ++i) {
                auto& index = tab_.indexes[i];
                auto ih = sm_manager_->ihs_.at(index.index_name).get();
                auto insert_key = index.get_key(rec.data);
                ih->insert_entry(insert_key.get(), rid, context_->txn_);
            }

        }
        std::chrono::high_resolution_clock::time_point end = std::chrono::high_resolution_clock::now();
        double total_time = std::chrono::duration_cast<std::chrono::duration<double>>(end - start).count();
        std::cout << "total_time = "<< total_time<< std::endl;
    }


    Rid &rid() override { return rid_; }

};